package com.intct.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-11-13 08:33
 */
public class MysqlCdcWatermark {
    public static void main(String[] args) {
        // 1. 初始化流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000L, org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE);

        // 2. 初始化表执行环境（适配 Flink 1.15+）
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        String createCdcSource = "CREATE TABLE mysql_t1_cdc (" +
                "  id STRING," +
                "  name STRING," +
                "  create_time TIMESTAMP(3)," +
                "  WATERMARK FOR create_time AS create_time - INTERVAL '5' SECOND," +
                "  PRIMARY KEY (id) NOT ENFORCED" +
                ") WITH (" +
                "  'connector' = 'mysql-cdc'," +
                "  'hostname' = 'cdh-node'," +
                "  'port' = '13306'," +
                "  'username' = 'root'," +
                "  'password' = 'Test_090110'," +
                "  'database-name' = 'm1'," +
                "  'table-name' = 'test01'," +
                "  'scan.startup.mode' = 'initial'" +
                ")";
        tableEnv.executeSql(createCdcSource);

        tableEnv.executeSql("select * from mysql_t1_cdc").print();
    }
}